package com.hex.ds.hdtp.core.inf.common.utils;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.util.StrUtil;
import com.hex.ds.hdtp.core.inf.common.enums.AuthType;
import com.hex.ds.hdtp.core.inf.common.enums.JdbcKerberosAuthConfigurationEnum;
import com.hex.ds.hdtp.core.inf.common.enums.JdbcKerberosSystemConfigurationEnum;
import com.hex.ds.hdtp.core.inf.common.enums.SourceJdbcErrorCode;
import com.hex.ds.hdtp.core.inf.common.exception.SourceJDBCException;
import com.hex.ds.hdtp.core.inf.metadata.po.SourceConfInfoPo;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.tomcat.jdbc.pool.DataSource;
import org.apache.tomcat.jdbc.pool.PoolProperties;
import sun.security.krb5.Config;
import sun.security.krb5.KrbException;

import java.sql.*;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

@Slf4j
public class JdbcUtil {

    private static final int POOL_PROPERTIES_INITIAL_SIZE = 1;
    public static final String KERBEROS_URL_PREFIX = ";principal=";

    public static final String KEY_TAB_FILE_PATH = "keyTabFilePath";

    public static final String AUTH_TYPE = "authType";

    public static final String PRINCIPAL_USER = "principalUser";

    /**
     * @Description: 测试连接，这个方法和下面getConnection很像，只不过添加了关闭连接，是为了节约开发中的冗余关闭连接代码
     * @Method: testConnection
     * @Param: sourceInfo 1
     * @Return: java.lang.Object
     * @Author: xiaoma
     * @Date: 2021/7/19 23:07
     * @Version: V1.0
     */
    public static Boolean testJdbcConnection(List<SourceConfInfoPo> sourceConfInfoPoList) {
        Connection jdbcConnection = null;
        DataSource jdbcDataSource = new DataSource();
        try {
            jdbcDataSource = getJdbcDataSource(sourceConfInfoPoList);
            jdbcConnection = jdbcDataSource.getConnection();
            if (null == jdbcConnection) {
                throw new SourceJDBCException(SourceJdbcErrorCode.ERROR_CODE_00006);
            }
        } catch (Exception exception) {
            log.error("get database connect failed,url:{}", jdbcDataSource.getUrl(), exception);
            throw new SourceJDBCException(SourceJdbcErrorCode.ERROR_CODE_00006, exception);
        } finally {
            jdbcDataSource.close();
            JdbcUtil.closeConnection(jdbcConnection);
        }
        log.info("get database connect success", jdbcConnection);
        return true;
    }

    /**
     * @Description: 获取数据源连接
     * @Method: getJdbcConnection
     * @Param: sourceInfo 1 数据源信息
     * @Return: java.sql.Connection
     * @Author: xiaoma
     * @Date: 2021/7/13 2:09
     * @Version: V1.0
     */
    public static Connection getJdbcConnection(List<SourceConfInfoPo> sourceConfInfoPoList) {
        Connection jdbcConnection = null;
        DataSource jdbcDataSource = new DataSource();
        try {
            jdbcDataSource = getJdbcDataSource(sourceConfInfoPoList);
            jdbcConnection = jdbcDataSource.getConnection();
            if (null == jdbcConnection) {
                throw new SourceJDBCException(SourceJdbcErrorCode.ERROR_CODE_00006);
            }
        } catch (Exception exception) {
            log.error("get database connect failed", exception);
            throw new SourceJDBCException(SourceJdbcErrorCode.ERROR_CODE_00006, exception);
        } finally {
            jdbcDataSource.close();
        }
        log.info("get database connect success", jdbcConnection);
        return jdbcConnection;
    }

    /**
     * @Description: 获取JDBC数据源信息
     * @Method: getJdbcDataSource
     * @Param: sourceInfo 数据源信息
     * @Return: org.apache.tomcat.jdbc.pool.DataSource
     * @Author: xiaoma
     * @Date: 2021/7/12 18:48
     * @Version: V1.0
     */
    public static DataSource getJdbcDataSource(List<SourceConfInfoPo> sourceConfInfoPoList) {
        //1 获取dataSource
        DataSource dataSource = getJdbcBasicInformation(sourceConfInfoPoList);
        //2 进行系统认证
        systemSecurityCertification(sourceConfInfoPoList);
        return dataSource;
    }

    private static void systemSecurityCertification(List<SourceConfInfoPo> sourceConfInfoPoList) {
        sourceConfInfoPoList = sourceConfInfoPoList.stream().filter(sourceConf -> StrUtil.isNotBlank(sourceConf.getConfValue())).collect(Collectors.toList());
        Map<String, String> sourceConfInfo = sourceConfInfoPoList.stream().collect(Collectors.toMap(SourceConfInfoPo::getConfName, SourceConfInfoPo::getConfValue));
        //1 系统环境变量设置
        SystemSecurityCertificationUtils.systemVariablesSetting(BeanUtil.beanToMap(sourceConfInfo), JdbcKerberosSystemConfigurationEnum.getValues());
        //2 系统用户登录
        systemUserLogin(sourceConfInfo);
    }

    private static void systemUserLogin(Map<String, String> sourceConfInfo) {
        if (AuthType.KERBEROS.getValue().equalsIgnoreCase(sourceConfInfo.get(AUTH_TYPE))) {
            Configuration configuration = getSystemUserLoginConfig(BeanUtil.beanToMap(sourceConfInfo));
            SystemSecurityCertificationUtils.systemUserLoginWithKeyTab(configuration, sourceConfInfo.get(PRINCIPAL_USER), sourceConfInfo.get(KEY_TAB_FILE_PATH));
        }
    }

    private static Configuration getSystemUserLoginConfig(Map<String, ?> paramMap) {
        Configuration configuration = new Configuration();
        List<String> jdbcKerberosSystemConfigurationKeyValues = JdbcKerberosAuthConfigurationEnum.getValues();
        List<String> paramKeys = new ArrayList<>(paramMap.keySet());
        configuration.set(JdbcKerberosAuthConfigurationEnum.HADOOP_SECURITY_AUTHENTICATION.getValue(), AuthType.KERBEROS.getValue());
        jdbcKerberosSystemConfigurationKeyValues.stream().forEach(kerberosKey -> {
            paramKeys.stream().forEach(paramKey -> {
                if (StringUtil.stringEqualsIgnoreCaseAndSymbol(paramKey, kerberosKey)) {
                    configuration.set(kerberosKey, paramMap.get(paramKey).toString());
                }
            });
        });
        return configuration;
    }

    private static DataSource getJdbcBasicInformation(List<SourceConfInfoPo> sourceConfInfoPoList) {
        PoolProperties poolProperties = getPoolProperties();
        DataSource dataSource = getDataSourceInformation(sourceConfInfoPoList, poolProperties);
        return dataSource;
    }

    private static PoolProperties getPoolProperties() {
        PoolProperties properties = new PoolProperties();
        properties.setInitialSize(POOL_PROPERTIES_INITIAL_SIZE);
        return properties;
    }

    private static DataSource getDataSourceInformation(List<SourceConfInfoPo> sourceConfInfoPoList, PoolProperties poolProperties) {
        sourceConfInfoPoList = sourceConfInfoPoList.stream().filter(sourceConf -> StrUtil.isNotBlank(sourceConf.getConfValue())).collect(Collectors.toList());
        Map<String, String> sourceConfInfo = sourceConfInfoPoList.stream().collect(Collectors.toMap(SourceConfInfoPo::getConfName, SourceConfInfoPo::getConfValue));
        sourceConfInfo.put("sourceType", sourceConfInfoPoList.get(0).getSourceType());
        DataSource dataSource = new DataSource(poolProperties);
        dataSource.setDriverClassName(sourceConfInfo.get("driverClassName"));
        dataSource.setUrl(getJDBCUrl(sourceConfInfo));
        setJDBCUsernameAndPassword(sourceConfInfo, dataSource);
        return dataSource;
    }

    /**
     * @Description: 获取基础URL
     * @Method: getBaseUrl
     * @Param: sourceJDBCEntity 1
     * @Param: dataSource 2
     * @Return: java.lang.String
     * @Author: xiaoma
     * @Date: 2021/11/25 16:08
     * @Version: V1.0
     */
    public static String getJDBCUrl(Map<String, String> sourceConfInfo) {
        String url = sourceConfInfo.get("url");
        // 处理数据源存在认证时（Kerberos、ldap等）URL信息调整
        url = handleUrlAuthInfo(sourceConfInfo, url);
        // 如果url中包含"odps"，根据以下方式进行相应的拼接
        if (url.contains("odps")) {
            if (url.contains("${endpoint}")) {
                url = url.replace("${endpoint}", sourceConfInfo.get("endpoint"));
            }
            if (url.contains("${project}")) {
                url = url.replace("${project}", sourceConfInfo.get("project"));
            }
        }
        if (url.contains("${host}")) {
            url = url.replace("${host}", sourceConfInfo.get("host"));
        }
        if (url.contains("${port}")) {
            url = url.replace("${port}", sourceConfInfo.get("port"));
        }
        if (url.contains("${serviceName}")) {
            url = url.replace("${serviceName}", sourceConfInfo.get("serviceName"));
        }
        if (url.contains("${tdEncode}")) {
            url = url.replace("${tdEncode}", sourceConfInfo.get("tdEncode"));
        }
        if (url.contains("${database}")) {
            url = url.replace("${database}", sourceConfInfo.get("database"));
        }
        return url;
    }

    /**
     * @Method handleUrlAuthInfo
     * @Param sourceConfInfo
     * @Param url
     * @Return java.lang.String
     * @Description 处理数据源存在认证时（Kerberos、ldap等）URL信息调整
     * @Author gj.xu
     * @Date 2024/2/22 16:50
     * @Version V2.0
     */
    private static String handleUrlAuthInfo(Map<String, String> sourceConfInfo, String url) {
        if (AuthType.KERBEROS.getValue().equalsIgnoreCase(sourceConfInfo.get(AUTH_TYPE))) {
            url = url.endsWith(StrUtil.SLASH) ? url : (url + StrUtil.SLASH);
            url += KERBEROS_URL_PREFIX + sourceConfInfo.get(PRINCIPAL_USER);
            verifyKerberos(sourceConfInfo);
        }
        // 如果url中包含"impala"，根据authType认证方式进行相应的拼接
        if ("IMPALA".equals(sourceConfInfo.get("sourceType")) || url.contains("impala")) {
            String authType = sourceConfInfo.get(AUTH_TYPE);
            if (AuthType.LDAP.getValue().equalsIgnoreCase(authType)) {
                url += ";AuthMech=3";
            } else if (AuthType.KERBEROS.getValue().equalsIgnoreCase(authType)) {
                url += ";AuthMech=1";
            } else if (AuthType.NONE.getValue().equalsIgnoreCase(authType)) {
                url += "/;auth=noSasl";
            }
        }
        return url;
    }

    /**
     * @Method verifyKerberos
     * @Param sourceConfInfo
     * @Return void
     * @Description 获取Kerberos认证信息
     * @Author Yin.Yang
     * @Date 2023/12/25 19:48
     * @Version V1.0
     */
    public static void verifyKerberos(Map<String, String> sourceConfInfo) {
        String kerberosKrb5Conf = sourceConfInfo.get("javaSecurityKrb5Conf");
        String kerberosJaas = sourceConfInfo.get("keyJaasPath");
        String kerberosKeytab = sourceConfInfo.get(KEY_TAB_FILE_PATH);
        String kerberosPrincipal = sourceConfInfo.get(PRINCIPAL_USER);
        // 设置krb5配置
        if (StrUtil.isNotBlank(kerberosKrb5Conf) && StrUtil.isEmpty(System.getProperty("java.security.krb5.conf"))) {
            System.setProperty("java.security.krb5.conf", kerberosKrb5Conf);
            try {
                Config.refresh();
            } catch (KrbException e) {
                throw new RuntimeException("Failed to refresh Kerberos configuration", e);
            }
        }
        // 设置JAAS配置
        if (StrUtil.isNotBlank(kerberosJaas) && StrUtil.isEmpty(System.getProperty("java.security.auth.login.config"))) {
            System.setProperty("java.security.auth.login.config", kerberosJaas);
        }
        // 设置其他系统属性
        System.setProperty("javax.security.auth.useSubjectCredsOnly", "false");
        System.setProperty("ipc.client.fallback-to-simple-auth-allowed", "true");
        // 使用Hadoop的UserGroupInformation进行Kerberos登录
        if (StrUtil.isNotBlank(kerberosPrincipal) && StrUtil.isNotBlank(kerberosKeytab)) {
            try {
                Configuration configuration = new Configuration();
                configuration.set("hadoop.security.authentication", "kerberos");
                configuration.set("hadoop.kerberos.keytab.login.autorenewal.enabled", "true");
                configuration.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");
                UserGroupInformation.setConfiguration(configuration);
                UserGroupInformation.loginUserFromKeytab(kerberosPrincipal, kerberosKeytab);
                log.info("当前用户：" + UserGroupInformation.getCurrentUser());
            } catch (Exception e) {
                throw new RuntimeException("Failed to authenticate with Kerberos", e);
            }
        }
    }


    /**
     * @Description: 获取JDBC连接的用户名和密码
     * @Method: getJDBCUsernameAndPassword
     * @Param: sourceConfInfo 1
     * @Param: dataSource 2
     * @Return: void
     * @Author: xiaoma
     * @Date: 2021/11/24 16:32
     * @Version: V1.0
     */
    private static void setJDBCUsernameAndPassword(Map<String, String> sourceConfInfo, DataSource dataSource) {
        if (!AuthType.KERBEROS.getValue().equalsIgnoreCase(sourceConfInfo.get(AUTH_TYPE))) {
            if (null != sourceConfInfo.get("username")) {
                dataSource.setUsername(sourceConfInfo.get("username"));
            }
            if (null != sourceConfInfo.get("password")) {
                dataSource.setPassword(sourceConfInfo.get("password"));
            }
        }
    }

    /**
     * @Description: 创建执行语句Statement
     * @Method: createJdbcStatement
     * @Param: jdbcConnection 1 数据源连接信息
     * @Return: java.sql.Statement
     * @Author: xiaoma
     * @Date: 2021/7/13 2:10
     * @Version: V1.0
     */
    public static Statement createJdbcStatement(Connection jdbcConnection) {
        Statement jdbcStatement = null;
        try {
            jdbcStatement = jdbcConnection.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
//            if (null == jdbcStatement) {
//                throw new SourceJDBCException(SourceJdbcErrorCode.ERROR_CODE_00008);
//            }
            // 如果驱动不是Hive.v1.1.0，设置SQL超时时间为8小时
            String driverClassName = jdbcConnection.getMetaData().getClass().getName();
            if (!driverClassName.equals("com.hex.hdtp.shaded.hive.v110.jdbc.HiveDatabaseMetaData")) {
                 jdbcStatement.setQueryTimeout(28800);
            }
        } catch (SQLException exception) {
            log.error("create jdbc statement failed", exception);
            throw new SourceJDBCException(SourceJdbcErrorCode.ERROR_CODE_00008, exception);
        }
        log.info("create jdbc statement success,result:{}", jdbcStatement);
        return jdbcStatement;
    }

    /**
     * @Description: 关闭数据源连接和Statement流和r结果集
     * @Method: closeConnection
     * @Param: connection 1 数据源连接信息
     * @Param: statement 2 Statement流对象
     * @Param: resultSet 3 结果集对象
     * @Return: void
     * @Author: xiaoma
     * @Date: 2021/7/20 11:48
     * @Version: V1.0
     */
    public static void closeConnection(Connection connection, Statement statement, ResultSet resultSet) {
        closeConnection(resultSet);
        closeConnection(statement);
        closeConnection(connection);
    }

    /**
     * @Description: 关闭数据源连接和Statement流
     * @Method: closeConnection
     * @Param: connection 1 数据源连接信息
     * @Param: statement 2 Statement流对象
     * @Return: void
     * @Author: xiaoma
     * @Date: 2021/7/13 14:12
     * @Version: V1.0
     */
    public static void closeConnection(Connection connection, Statement statement) {
        closeConnection(statement);
        closeConnection(connection);
    }

    /**
     * @Description: 关闭数据源连接
     * @Method: closeConnection
     * @Param: connection 1 数据源连接信息
     * @Return: void
     * @Author: xiaoma
     * @Date: 2021/7/13 9:28
     * @Version: V1.0
     */
    public static void closeConnection(Connection connection) {
        if (null != connection) {
            try {
                connection.close();
                log.info("close jdbc connection success,connection:{}", connection);
            } catch (SQLException exception) {
                log.error("close jdbc connection failed", exception);
                throw new SourceJDBCException(SourceJdbcErrorCode.ERROR_CODE_00006, exception);
            }
        }
    }

    /**
     * @Description: 关闭数据源连接
     * @Method: closeConnection
     * @Param: statement 1 Statement流对象
     * @Return: void
     * @Author: xiaoma
     * @Date: 2021/7/20 11:52
     * @Version: V1.0
     */
    public static void closeConnection(Statement statement) {
        if (null != statement) {
            try {
                statement.close();
                log.info("close jdbc statement success,statement:{}", statement);
            } catch (SQLException exception) {
                log.error("close jdbc statement failed", exception);
                throw new SourceJDBCException(SourceJdbcErrorCode.ERROR_CODE_00006, exception);
            }
        }
    }

    /**
     * @Description: 关闭数据源连接
     * @Method: closeConnection
     * @Param: resultSet 1 结果集对象
     * @Return: void
     * @Author: xiaoma
     * @Date: 2021/7/20 11:53
     * @Version: V1.0
     */
    public static void closeConnection(ResultSet resultSet) {
        if (null != resultSet) {
            try {
                resultSet.close();
                log.info("close jdbc statement success,resultSet:{}", resultSet);
            } catch (SQLException exception) {
                log.error("close jdbc statement failed", exception);
                throw new SourceJDBCException(SourceJdbcErrorCode.ERROR_CODE_00006, exception);
            }
        }
    }

    /**
     * @Description: 格式化SQL，去掉SQL前后的空格和最后后的分号
     * @Method: formatSql
     * @Param: sql 1 sql语句
     * @Return: java.lang.String
     * @Author: xiaoma
     * @Date: 2021/7/15 11:00
     * @Version: V1.0
     */
    public static String formatSql(String sql) {
        return StrUtil.removeSuffix(StrUtil.trimToEmpty(sql), ";");
    }

    /**
     * @Description: 执行Select类型的sql
     * @Method: executeSelectSql
     * @Param: jdbcStatement 1
     * @Param: sql 2
     * @Param: jdbcUrl 3
     * @Param: exceptionSql 4
     * @Return: java.sql.ResultSet
     * @Author: xiaoma
     * @Date: 2022/5/11 16:05
     * @Version: V1.0
     */
    public static ResultSet executeQuerySql(Statement jdbcStatement, String querySql) {
        ResultSet resultSet = null;
        try {
            String formatSql = JdbcUtil.formatSql(querySql);
            log.info(StrUtil.format("执行SQL：{}", formatSql));
            resultSet = jdbcStatement.executeQuery(formatSql);
        } catch (SQLException exception) {
            throw new SourceJDBCException(SourceJdbcErrorCode.ERROR_CODE_00013, exception);
        }
        return resultSet;
    }


    /**
     * @Description: 执行非Select类型的sql
     * @Method: executeNonSelectSql
     * @Param: jdbcStatement 1
     * @Param: sql 2
     * @Param: jdbcUrl 3
     * @Param: exceptionSql 4
     * @Return: java.lang.Integer
     * @Author: xiaoma
     * @Date: 2022/5/11 16:06
     * @Version: V1.0
     */
    public static void executeUpdatetSql(Statement jdbcStatement, String updateSql) {
        try {
            log.info(StrUtil.format("执行SQL：{}", updateSql));
            jdbcStatement.execute(JdbcUtil.formatSql(updateSql));
        } catch (SQLException exception) {
            throw new SourceJDBCException(SourceJdbcErrorCode.ERROR_CODE_00014, exception);
        }
    }
}
